<?php

namespace job;

use AMQPChannel;
use AMQPConnection;
use AMQPConnectionException;
use AMQPExchange;
use AMQPQueue;
use app\admin\model\base\SysLog;
use app\datacenter\service\LargeScreenService;
use think\exception\ErrorException;
use think\facade\Log;

/**
 * RabbitmqJob队列
 */
class RabbitmqJob
{
    //连接参数
    public string $rabbitmq_host;
    public int $rabbitmq_port;
    public string $rabbitmq_login;
    public string $rabbitmq_password;
    //Rabbitmq
    protected AMQPConnection $connection;
    protected AMQPChannel $channel;
    protected string $exchangeName;
    protected string $routeKey;
    //保存实例的对象
    private static $_instance;

    /**
     * 定义构造方法
     * @throws AMQPConnectionException
     */
    private function __construct()
    {
        //根据环境获取参数
        switch (env('ENVIRONMENT')) {
            case 'dev':
                $this->rabbitmq_host = '106.55.164.145';
                $this->rabbitmq_port = 5672;
                $this->rabbitmq_login = 'wusudiqiu';
                $this->rabbitmq_password = 'wusudiqiu';
                break;
            default:
                $this->rabbitmq_host = '120.24.206.44';
                $this->rabbitmq_port = 5672;
                $this->rabbitmq_login = 'wsdq2021';
                $this->rabbitmq_password = 'wsdq2021';
                break;
        }
        //Rabbitmq
        $this->exchangeName = 'wsdq_emqx_exchange';
        $this->routeKey = 'wsdq_emqx_route';
        $this->connection = new AMQPConnection([
            'host' => $this->rabbitmq_host,
            'port' => $this->rabbitmq_port,
            'vhost' => '/',
            'login' => $this->rabbitmq_login,
            'password' => $this->rabbitmq_password,
            //心跳检测
            'heartbeat' => 60,
        ]);
        //创建连接和channel
        $this->connection->connect() or die("Cannot connect to the broker!\n");
        $this->channel = new AMQPChannel($this->connection);
    }

    //定义一个空方法防止被外部克隆
    private function __clone()
    {
    }

    /**
     * 定义一个获取实例对象的方法
     * @return RabbitmqJob
     */
    public static function getInstance(): RabbitmqJob
    {
        if (!(self::$_instance instanceof RabbitmqJob)) {
            self::$_instance = new self();
        }
        //返回实例对象
        return self::$_instance;
    }

    /**
     * 发送消息至Rabbitmq
     * @param string $type
     * @param string $topic
     * @param mixed $msg
     * @return bool|null
     */
    public function sendMsg(string $type, string $topic, mixed $msg): bool|null
    {
        $data = [
            'type' => $type,
            'topic' => $topic,
            'msg' => $msg,
            'time' => time(),
        ];
        try {
            //创建交换机对象
            $exchange = new AMQPExchange($this->channel);
            $exchange->setName($this->exchangeName);
            $exchange->setType(AMQP_EX_TYPE_DIRECT);
            $exchange->declareExchange();
            $exchange->publish(json_encode($data), $this->routeKey);
            return true;
        } catch (AMQPConnectionException $e) {
            Log::write($e->getMessage(), '发送消息至Rabbitmq');
        }
        //断开连接
        $this->connection->disconnect();
        return false;
    }

    /**
     * RabbitmqEmqx消费者
     * @param string $queueName
     * @return mixed
     */
    public function receive(string $queueName): mixed
    {
        $exchange = new AMQPExchange($this->channel);
        $exchange->setName($this->exchangeName);
        $exchange->setType(AMQP_EX_TYPE_DIRECT);
        $Temp['Exchange Status'] = $exchange->declareExchange();
        ################################################################
        $queue = new AMQPQueue($this->channel);
        $queue->setName($queueName);
        $Temp['Message Total'] = $queue->declareQueue();
        $queue->bind($this->exchangeName, $this->routeKey);
        var_dump($Temp);
        // 消费队列消息
        while (TRUE) {
            $queue->consume(function ($envelope, $queue) {
                $data = $envelope->getBody();
                //执行消费程序
                self::consumerProgram(json_decode($data, true));
                //手动发送ACK应答
                $queue->ack($envelope->getDeliveryTag());
            });
        }//while
    }

    /**
     * 执行消费程序分发窗口
     * @param array $data
     * @return void
     */
    private static function consumerProgram(array $data)
    {
        //根据类型走不同流程
        switch ($data['type']) {
            //写入系统日志
            case 'WriteSystemLog':
                SysLog::writeSysLog($data['msg']);
                break;
            //更新系统日志
            case 'UpdateSystemLog':
                SysLog::updateSysLog($data['msg']);
                break;
            //微信取袋推送至数据中心大屏
            case 'V1PushToDataScreen':
                try {
                    $LargeScreenService = new LargeScreenService();
                    $LargeScreenService->getMmberDataOne($data['msg']);
                } catch (\Exception|ErrorException $exception) {
                    Log::write($exception, '微信取袋推送至数据中心大屏');
                }
                break;
            default:
                break;
        }
    }

}